<?php declare(strict_types=1);


namespace App\Rpc\Service;
use Swoft\Redis\Pool;
use App\Rpc\Lib\MessageInterface;
use Swoft\Bean\Annotation\Mapping\Inject;
use Swoft\Rpc\Server\Annotation\Mapping\Service;

/**
 * Class MessageService
 *
 * @since 2.0
 *
 * @Service()
 */
class MessageService implements MessageInterface
{

    /**
     * @Inject("redis.pool")
     * @var Pool
     */
    private $connectionRedis;

    /**
     * @Inject("rabbit.pool")
     * @var \Six\Rabbit\Pool
     */
    private $rabbit;
    /**
     * 预发送消息（消息预发送子系统）
     * @param $prepareMsgData
     * @return array
     */
    public function prepareMsg($prepareMsgData): array
    {

        $msg_id = $prepareMsgData['msg_id'];

        #存储数据，还要存储时间，依据时间查找超时的任务
        $result = $this->connectionRedis->transaction(function (\Redis $redis) use ($msg_id,$prepareMsgData){
            $redis->hset("message_system", (string)$msg_id, json_encode($prepareMsgData));
            $redis->zAdd("message_system_time",$prepareMsgData['create_time'],(string)$msg_id);
        });
//        var_dump('redis事务',$result);
        if ($result[0] == false) {
            return ['status' => 0, 'result' => '预发送消息失败'];
        }
        return ['status' => 1, 'result' => '预发送消息成功'];
    }

    /**
     * 确认并且投递参数（消息确认投递子系统）
     * @param $msg_id
     * @param $flag
     * @return array
     */
    public function confirmMsgToSend($msg_id ,$flag):array
    {
        try{
            $connection = $this->rabbit->connect();
            $connectionRabbit = $connection->connection;

            $exchangeName = "tradeExchange"; #通道名称
            $routeKey = "/trade";
            $queueName = 'trade'; //队列名称
            #连接创建通道
            $channel = $connectionRabbit->channel();
            /**
             * 创建队列(Queue)
             * name: hello         // 队列名称
             * passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建
             * durable: true       // 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失
             * exclusive: false    // 是否排他，指定该选项为true则队列只对当前连接有效，连接断开后自动删除
             *  auto_delete: false // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
             */
            $channel->queue_declare($queueName, false, true, false, false);
            #建立交换机通过通道去通讯

            /**
             * 创建交换机(Exchange)
             * name: vckai_exchange// 交换机名称
             * type: direct        // 交换机类型，分别为direct/fanout/topic，参考另外文章的Exchange Type说明。
             * passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建
             * durable: false      // 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失
             * auto_delete: false  // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
             */
            $channel->exchange_declare($exchangeName, \PhpAmqpLib\Exchange\AMQPExchangeType::DIRECT, false, true, false);
//            var_dump(1111);
            // 绑定消息交换机和队列
            $channel->queue_bind($queueName, $exchangeName,$routeKey);
            $data = $this->connectionRedis->hget("message_system", (string)$msg_id);

            if(!empty($data)){
                $data=json_decode($data,true);
                $data['status']=2;

                if($flag==2){
                    var_dump('被消息恢复子系统投递的任务');

                    $data['message_retries_number']=$data['message_retries_number']+1;//被消息恢复子系统投递的任务
                }
                $data=json_encode($data);

                /**
                 * 创建AMQP消息类型
                 * delivery_mode 消息是否持久化
                 * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT  不持久化
                 * AMQPMessage::DELIVERY_MODE_PERSISTENT      持久化
                 */
                $msg = new \PhpAmqpLib\Message\AMQPMessage($data, ['delivery_mode' => \PhpAmqpLib\Message\AMQPMessage:: DELIVERY_MODE_NON_PERSISTENT]);

                //发布消息到交换机当中,并且绑定好路由关系
                if($this->connectionRedis->hset("message_system",(string)$msg_id,$data)==0 && $channel->basic_publish($msg,$exchangeName, $routeKey) == null ){
                    //将消息投递给MQ(实时消息服务)
                    $result = ['status' =>1 ,'result' => '确认并且投递成功'];
                }else{
                    $result = ['status' =>0 ,'result' => '确认并且投递失败'];
                }
            }else{
                $result = ['status' =>0 ,'result' => '确认并且投递失败'];
            }
            $channel->close();
            $connection->release(true); #放入连接池，重用
            return $result;
        }catch (\Exception $exception){
            return ['message' =>$exception->getMessage(),'line' => $exception->getLine() ,'code' =>$exception->getCode()];
        }

    }


    /**
     * 消息消费成功（消息确认子系统）
     * @param $msg_id
     * @return array
     * @throws \ReflectionException
     * @throws \Swoft\Bean\Exception\ContainerException
     */
    public function ackMsg($msg_id):array
    {
        $connectionRedis = bean('redis.pool')->getConnection();
        $connectionRedis->setRelease(true);

        #存储数据，还要存储事件，依据时间查找超时的任务
        $result = $connectionRedis->transaction(function (\Redis $redis) use ($msg_id){
            $redis->hDel("message_system", (string)$msg_id);
            $redis->zRem("message_system_time",(string)$msg_id);
        });
        var_dump('消息确认成功',$result);
        $connectionRedis->release();

        return ['status' =>1 ,'result' => '任务消费成功'];
    }

    /**
     * 消息确认状态（消息）
     * @param $msgType
     * @return array
     */
    public function selectMsgTime($msgType):array
    {
        return ['status' =>1 ,'result' => '查询到的任务状态'];
    }
}